package cug.example.learningbase.FileUtils;

import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.io.File;

import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

@Component
public class FilePushManager {
    //创建一个并发集合，用于记录已提交或已处理的文件名
    private final Set<String> processedFiles = ConcurrentHashMap.newKeySet();
    private final BlockingQueue<File> fileQueue = new LinkedBlockingQueue<>();
    private final int THREAD_COUNT = 2;
    private final ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT);
    private final String FILE_PATH = "D:\\01\\test";
    private volatile AtomicBoolean started = new AtomicBoolean(false);//原子操作：保证多个线程同时调用 start() 也只会启动一次；
    @Scheduled(fixedRate = 1000)
    public void start() {
        // 启动生产者线程（扫描文件夹并放入队列）
        scanAndEnqueueFiles();
        // 启动消费者线程（worker）
        if(started.compareAndSet(false, true)){
            for (int i = 0; i < THREAD_COUNT; i++) {
                executor.submit(this::consumeFiles);
            }
        }
    }

    // 生产者：扫描文件夹，将文件放入队列
    private void scanAndEnqueueFiles() {
        File directory = new File(FILE_PATH);
        File[] files = directory.listFiles();
        if (files != null) {
            for (File file : files) {
                String name = file.getName();
                if (processedFiles.contains(name)) continue; // 已处理或正在处理，跳过
                processedFiles.add(name); // 标记为处理中
                try {
                    fileQueue.put(file); // put() 是阻塞式，线程安全
                    System.out.println("放入队列：" + file.getName());
                } catch (InterruptedException e) {
                    processedFiles.remove(name);
                    Thread.currentThread().interrupt();
                }
            }
        }
    }

    // 消费者：不断从队列中拿文件并处理
    private void consumeFiles() {
        while (true) {
            try {
                File file = fileQueue.take(); // take() 会阻塞直到拿到
                System.out.println(Thread.currentThread().getName() + " 处理文件：" + file.getName());
                processFile(file);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }
    }

    // 文件处理逻辑（基本保留你的逻辑）
    private void processFile(File file) {
        System.out.println(file.getName());
        try{
            //文件处理逻辑
        }catch(Exception e){

        }finally {
            processedFiles.remove(file.getName());// 移除文件名
        }
        deleteFolder(file);
    }

    // 文件夹递归删除
    private void deleteFolder(File file) {
        if (file.isDirectory()) {
            for (File child : file.listFiles()) {
                deleteFolder(child);
            }
        }
        file.delete();
    }


    public static void main(String[] args) {

        FilePushManager filePushManager = new FilePushManager();
        filePushManager.start();
    }
}
